package com.shujia.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object Demo2AggregateByKey {
  def main(args: Array[String]): Unit = {
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    val sparkContext: SparkContext = sparkSession.sparkContext

    val studentsRDD: RDD[String] = sparkContext.textFile("spark/data/students.csv")

    val clazzKVRDD: RDD[(String, Int)] = studentsRDD.map((line: String) => (line.split(",")(4), 1))

    /**
     * reduceByKey: 分组聚合，但是只会在reduce端进行聚合
     *
     *
     * AggregateByKey: 分组聚合，不仅可以设置reduce端聚合的方式，可以提前的在map端进行聚合，相当于预聚合
     * zeroValue：初始值
     * seqOp: (U, V) => U：map端的聚合
     * combOp: (U, U) => U：reduce端的聚合
     *
     * 如果当map聚合逻辑和reduce聚合逻辑不一样的时候，需要分别设置的时候，就可以使用该算子
     *
     */
    //aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]
//    clazzKVRDD.aggregateByKey(0)(
//      (u: Int, v: Int) => u + v, // 在map端做聚合操作的函数
//      (u1: Int, u2: Int) => u1 + u2 //在reduce端做聚合操作
//    ).foreach(println)

    //reduceByKey(func: (V, V) => V)
    clazzKVRDD.reduceByKey(_ + _).foreach(println)

  }
}
